消息队列 与 邮件发送

起因

  1. 消息队列顾名思义是发消息的，但是它并不是一个用来发消息功能；它的作用是用于线程与线程间，进程与进程间的通信；

消息队列

  1. 消息队列可认为是一种通用的解决方案，一种进程（线程）间通信的方案；有句话是这样讲的，不要以共享内存来实现通信，而应该通过通信来实现共享内存；当我们想把一件事情共享给多个进程，而又不需要强烈的时间限制，那么通信是一种很好的解决方案；两个进程如果都需要某个变量，那就通过通信传递，实现共享，协同工作；而消息队列是什么样的消息呢？它的通信实际上传递的是事件，或者工作；
  2. 消息队列的目的是为了将一些不那么需要及时响应的活传递给另一个线程或者进程；这样主线程或者进程就可以快速的处理事情；
  3. 举个不太恰当的例子，我是某办事处的老大，每天有n多投诉会到我们办事处，我会差不多看一遍，分分类，然后如果有红色等级就自己批注一下，否则的话，我就通通交给门口看门的街道办事处大爷；然后大爷就会找邻里街坊的把事情给处理掉；
  4. 特点是：
    1. 我很忙，主线程的运行不能搁置，就好像双11，大家总不能都挤在入口的服务器上，我其实做了负载均衡或者代理的作用，所有事情都往我这里投诉，然后我在分配下去；
    2. 异步性，这个处理并不是必须及时的，要是我的顶头上司过来投诉，我才不管别人，专门服务与我的上司；
    3. 投诉很多，假设每天就一个投诉，那占用我的全部时间也没关系，但是当投诉堆积成山，我就会陷入宕机的状态，无法处理；
  5. 一个消费者也可以是一个生产者，比如说快递吧，一件商品运到目的地需要很多手传递，我们完全可以认为这是一个流水线的过程，任何一个消费者都可能是下一步的生产者；

    代码

  • 类图描述
    这里写图片描述
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@RequestMapping(path = {"/like"}, method = {RequestMethod.POST})
    @ResponseBody
    public String like(@RequestParam("commentId") int commentId) {
        if (hostHolder.getUser() == null) {
            return WendaUtil.getJSONString(999);
        }
        Comment comment = commentService.getCommentById(commentId);
//添加的部分，用于添加事件
        eventProducer.fireEvent(new EventModel(EventType.LIKE)
                .setActorId(hostHolder.getUser().getId())
                .setEntityId(commentId)
                .setEntityType(EntityType.ENTITY_COMMENT)
                .setExt("questionId", String.valueOf(comment.getEntityId()))
                .setEntityOwnerId(comment.getUserId()));
        long likeCount = likeService.like(hostHolder.getUser().getId(), EntityType.ENTITY_COMMENT, commentId);
        return WendaUtil.getJSONString(0, String.valueOf(likeCount));
    }

  • 整个消费者

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    @Service
    public class EventConsumer  implements InitializingBean,ApplicationContextAware{
        private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);
        //某个EventType 所对应的Handler链表；
        private Map<EventType, List<EventHandler>> config = new HashMap<>();
        //上下文，可以通过它来找到所有实现EventHandler接口的类
        private ApplicationContext applicationContext;
        @Autowired
        JedisAdapter jedisAdapter;
        @Override
        public void afterPropertiesSet() throws Exception {
            Map<String, EventHandler> beans = applicationContext.getBeansOfType(EventHandler.class);
            if (beans != null) {
                for (Map.Entry<String, EventHandler> entry : beans.entrySet()) {
                    //对于某个handler，他关注那些事件类型
                    List<EventType> eventTypes = entry.getValue().getSupportEventTypes();
                    for (EventType eventType : eventTypes) {
                        //这个事件被第一次注册
                        if (!config.containsKey(eventType)) {
                            config.put(eventType, new ArrayList<>());
                        }
                        //将这个handler加入到他能处理的事件类型的那条链表中
                        config.get(eventType).add(entry.getValue());
                    }
                }
            }
            //新建一个线程，简单来说，就是一直循环，有事件过来就消费；
            Thread thread = new Thread(() -> {
                while (true) {
                    String key = RedisKeyUtil.getEventQueueKey();
                    List<String> events = jedisAdapter.brpop(0, key);
                    for (String message : events) {
                        if (message.equals(key)) {
                            continue;
                        }
                        // 还原event
                        EventModel eventModel = JSON.parseObject(message, EventModel.class);
                        if (!config.containsKey(eventModel.getType())) {
                            logger.error("非法事件");
                            continue;
                        }
                        //  对于该event所对应的所有handler，让handler进行处理
                        for (EventHandler eventHandler : config.get(eventModel.getType())) {
                            eventHandler.doHandle(eventModel);
                        }
                    }
                }
            });
            thread.start();
        }
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
        }
    }

  • 生产者

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    @Service
    public class EventProducer {
        @Autowired
        JedisAdapter jedisAdapter;
        /**
         * 添加时间给消息队列，这里消息队列由Redis实现
         * 由于这里并不存在优先级级的问题，所以就用普通链表即可
         * 这里将事件json话，然后使用kv存入redis，然后就可以从redis在读出来，
         * 还原成事件
         * @param eventModel
         * @return
         */
        public boolean fireEvent(EventModel eventModel) {
            try {
                String json = JSONObject.toJSONString(eventModel);
                String key = RedisKeyUtil.getEventQueueKey();
                jedisAdapter.lpush(key, json);
                return true;
            } catch (Exception e) {
                return false;
            }
        }
    }

邮件发送

  1. 这部分其实不难，就是写共有代码，问题实际就在ssl，它需要安全的连接，但是我自己的邮件服务器安全协议又会被拒绝，qq就没问题，不过需要使用授权码；
继续跑<br><br>带着赤子的骄傲